// Copyright 2018 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// ------------------------------------------------------------------

#include <TCPTransportInterface.h>
#include <transport/TransportFactory.hpp>
#include <deps/common/System.h>
#include <TCPSenderResource.hpp>
#include <TCPChannelResourceBasic.h>
#include <TCPAcceptorBasic.h>
#if TLS_FOUND
#include <TCPChannelResourceSecure.h>
#include <TCPAcceptorSecure.h>
#endif  // if TLS_FOUND
#if defined(_WIN32)
#include <winsock2.h>
#include <ws2tcpip.h>
#else
#include <arpa/inet.h>
#endif

#include <deps/common/IPLocator.h>

#include <utility>
#include <cstring>
#include <algorithm>
#include <chrono>
#include <thread>

#include <asio/steady_timer.hpp>
#include <elog/Log.hpp>
#include <ertps/utils/LocatorAlias.h>

using namespace std;
using namespace asio;

namespace vbs {
namespace transport {

using octet = ertps::rtps::octet;
using IPLocator = ertps::rtps::IPLocator;
using SenderResource = vbs::transport::SenderResource;
using PortParameters = vbsutil::xmlparser::PortParameters;

TCPTransportDescriptor::TCPTransportDescriptor()
    : SocketTransportDescriptor(s_maximumMessageSize, s_maximumInitialPeersRange),
      tcp_negotiation_timeout(0),
      enable_tcp_nodelay(false),
      wait_for_tcp_negotiation(false) {}

TCPTransportDescriptor::TCPTransportDescriptor(const TCPTransportDescriptor& t)
    : SocketTransportDescriptor(static_cast<const SocketTransportDescriptor&>(t)),
      listening_ports(t.listening_ports),
      rtcp_header_(t.rtcp_header_),
      tcp_negotiation_timeout(t.tcp_negotiation_timeout),
      enable_tcp_nodelay(t.enable_tcp_nodelay),
      wait_for_tcp_negotiation(t.wait_for_tcp_negotiation) {}

TCPTransportDescriptor& TCPTransportDescriptor::operator=(const TCPTransportDescriptor& t) {
    maxMessageSize = t.maxMessageSize;
    maxInitialPeersRange = t.maxInitialPeersRange;
    sendBufferSize = t.sendBufferSize;
    receiveBufferSize = t.receiveBufferSize;
    TTL = t.TTL;
    listening_ports = t.listening_ports;
    rtcp_header_ = t.rtcp_header_;
    tcp_negotiation_timeout = t.tcp_negotiation_timeout;
    enable_tcp_nodelay = t.enable_tcp_nodelay;
    wait_for_tcp_negotiation = t.wait_for_tcp_negotiation;
    return *this;
}

bool TCPTransportDescriptor::operator==(const TCPTransportDescriptor& t) const {
    return ((this->listening_ports == t.listening_ports) && (this->rtcp_header_ == t.rtcp_header_) &&
            (this->tcp_negotiation_timeout == t.tcp_negotiation_timeout) &&
            (this->enable_tcp_nodelay == t.enable_tcp_nodelay) &&
            (this->wait_for_tcp_negotiation == t.wait_for_tcp_negotiation) && SocketTransportDescriptor::operator==(t));
}

TCPTransportInterface::TCPTransportInterface(int32_t transport_kind)
    : TransportInterface(transport_kind), alive_(true) {}

TCPTransportInterface::~TCPTransportInterface() {}

void TCPTransportInterface::clean() {
    assert(receiver_resources_.size() == 0);
    alive_.store(false);
    logDebug(RTCP, "TCPTransportInterface clean alive_:" << alive_.load());
    {
        std::unique_lock<std::mutex> scopedLock(send_resources_mutex_);
        logDebug(RTCP, "Clean channel_resources_ send_resources_mutex_ get lock");
        for (auto it = channel_resources_.begin(); it != channel_resources_.end();) {
            it->second->disconnect();
            it->second->clear();

            it = channel_resources_.erase(it);
            logDebug(RTCP, "Clean channel_resources_ disconnect:" << IPLocator::to_string(it->first)
                                                                  << " success, channel_resources_ size:"
                                                                  << channel_resources_.size());
        }
        logDebug(RTCP, "Clean channel_resources_ send_resources_mutex_ unlock");
    }

    if (io_service_thread_) {
        io_service_.stop();
        io_service_thread_->join();
        io_service_thread_ = nullptr;
        logDebug(RTCP, "io_service_thread_ join success");
    }
}

bool TCPTransportInterface::create_acceptor_socket(const Locator& locator) {
    try {
        std::unique_lock<std::mutex> scopedLock(acceptors_mutex_);
        std::shared_ptr<TCPAcceptorBasic> acceptor = std::make_shared<TCPAcceptorBasic>(io_service_, this, locator);
        acceptors_[locator] = acceptor;
        acceptor->accept(this);

        logDebug(RTCP, " OpenAndBindInput (physical: " << IPLocator::getPhysicalPort(locator)
                                                       << "; logical: " << IPLocator::getLogicalPort(locator) << ")");
    } catch (asio::system_error const& e) {
        (void)e;
        logError(RTCP, "TCPTransport Error binding at port: (" << IPLocator::getPhysicalPort(locator) << ")"
                                                               << " with msg: " << e.what());
        return false;
    } catch (const asio::error_code& code) {
        (void)code;
        logError(RTCP, "TCPTransport Error binding at port: (" << IPLocator::getPhysicalPort(locator) << ")"
                                                               << " with code: " << code);
        return false;
    }

    return true;
}

bool TCPTransportInterface::init() {
    if ((configuration()->sendBufferSize == 0) || (configuration()->receiveBufferSize == 0)) {
        // Check system buffer sizes.
        ip::tcp::socket socket(io_service_);
        socket.open(generate_protocol());

        if (configuration()->sendBufferSize == 0) {
            socket_base::send_buffer_size option;
            socket.get_option(option);
            set_send_buffer_size(option.value());

            if (configuration()->sendBufferSize < s_minimumSocketBuffer) {
                set_send_buffer_size(s_minimumSocketBuffer);
            }
        }

        if (configuration()->receiveBufferSize == 0) {
            socket_base::receive_buffer_size option;
            socket.get_option(option);
            set_receive_buffer_size(option.value());

            if (configuration()->receiveBufferSize < s_minimumSocketBuffer) {
                set_receive_buffer_size(s_minimumSocketBuffer);
            }
        }

        socket.close();
    }

    if (configuration()->maxMessageSize > s_maximumMessageSize) {
        logError(RTCP, "maxMessageSize cannot be greater than 65000");
        return false;
    }

    if (configuration()->maxMessageSize > configuration()->sendBufferSize) {
        logError(RTCP, "maxMessageSize cannot be greater than send_buffer_size");
        return false;
    }

    if (configuration()->maxMessageSize > configuration()->receiveBufferSize) {
        logError(RTCP, "maxMessageSize cannot be greater than receive_buffer_size");
        return false;
    }

    // TODO(Ricardo) Create an event that update this list.
    get_ips(current_interfaces_);

    auto ioServiceFunction = [&]() {
#if ASIO_VERSION >= 101200
        asio::executor_work_guard<asio::io_service::executor_type> work(io_service_.get_executor());
#else
        io_service::work work(io_service_);
#endif  // if ASIO_VERSION >= 101200
        (void)io_service_.run();
    };
    io_service_thread_ = std::make_shared<std::thread>(ioServiceFunction);

    return true;
}

bool TCPTransportInterface::is_input_port_open(uint16_t port) const {
    std::unique_lock<std::mutex> scopedLock(receiver_resources_mutex_);
    return receiver_resources_.find(port) != receiver_resources_.end();
}

bool TCPTransportInterface::IsInputChannelOpen(const Locator& locator) const {
    return IsLocatorSupported(locator) && is_input_port_open(IPLocator::getPhysicalPort(locator));
}

bool TCPTransportInterface::IsLocatorSupported(const Locator& locator) const {
    return locator.kind == transport_kind_;
}

bool TCPTransportInterface::transform_remote_locator(const Locator& remote_locator, Locator& result_locator) const {
    (void)remote_locator;
    (void)result_locator;
    return true;
}

bool TCPTransportInterface::DoInputLocatorsMatch(const Locator& left, const Locator& right) const {
    return IPLocator::getPhysicalPort(left) == IPLocator::getPhysicalPort(right);
}
void TCPTransportInterface::CloseOutputChannel(std::shared_ptr<TCPChannelResource>& channel) {
    Locator physical_locator = channel->remote_locator();
    channel.reset();
    logDebug(RTCP, "CloseOutputChannel channel reset success");
    std::unique_lock<std::mutex> scopedLock(send_resources_mutex_);
    logDebug(RTCP, "CloseOutputChannel send_resources_mutex_ get lock success");

    auto channel_resource = channel_resources_.find(physical_locator);
    if (channel_resource != channel_resources_.end()) {
        channel_resource->second->disconnect();
        channel_resource->second->clear();
        logDebug(RTCP, "TCPChannelResource status:" << channel_resource->second->connection_status());
        channel_resources_.erase(channel_resource);
        logDebug(RTCP, "CloseOutputChannel:" << IPLocator::to_string(physical_locator) << " success");
    } else {
        logDebug(RTCP, "CloseOutputChannel:" << IPLocator::to_string(physical_locator) << " failed"
                                             << "channel resource size:" << channel_resources_.size());
        //遍历channel_resources_，并打印
        for (auto it = channel_resources_.begin(); it != channel_resources_.end(); ++it) {
            logDebug(RTCP, "CloseOutputChannel channel_resources_:" << IPLocator::to_string(it->first));
        }
    }
}

bool TCPTransportInterface::CloseInputChannel(const Locator& locator) {
    bool bClosed = false;
    {
        std::unique_lock<std::mutex> scopedLock(acceptors_mutex_);
        auto acceptor = acceptors_.find(locator);
        if (acceptor != acceptors_.end()) {
            // acceptor->second->cancel();
            // acceptor->second->close();
            acceptors_[locator] = nullptr;
            logDebug(RTCP, "CloseInputChannel acceptors_ erase:" << IPLocator::to_string(locator) << " success");
        }
    }

    {
        std::unique_lock<std::mutex> scopedLock(receiver_resources_mutex_);

        uint16_t physicalPort = IPLocator::getPhysicalPort(locator);
        auto receiverIt = receiver_resources_.find(physicalPort);
        if (receiverIt != receiver_resources_.end()) {
            bClosed = true;
            (void)receiver_resources_.erase(receiverIt);
            logDebug(RTCP,
                     "CloseInputChannel receiver_resources_ erase:" << IPLocator::to_string(locator) << " success");
        }
    }

    // TODO 清理channel_resources_
    return bClosed;
}

void TCPTransportInterface::close_tcp_socket(std::shared_ptr<TCPChannelResource>& channel) {
    channel->disconnect();

    channel->change_status(TCPChannelResource::eConnectionStatus::eDisconnected);
    // channel.reset(); lead to race conditions because TransportInterface functions used in the
    // callbacks doesn't check validity.
}

bool TCPTransportInterface::OpenOutputChannel(std::map<Locator_t, std::shared_ptr<SenderResource>>& send_resource_map,
                                              const Locator_t& remote_locator, Locator_t& local_locator) {
    if (!IsLocatorSupported(remote_locator)) {
        return false;
    }

    bool success = false;
    uint16_t physical_port = IPLocator::getPhysicalPort(remote_locator);

    if (physical_port != 0) {
        // We try to find a SenderResource that can be reuse to this locator.
        // Note: This is done in this level because if we do in NetworkFactory level, we have to
        // mantain what transport already reuses a SenderResource.
        auto sender_resource = send_resource_map.find(remote_locator);
        if (sender_resource != send_resource_map.end()) {
            TCPSenderResource* tcp_sender_resource = TCPSenderResource::cast(*this, sender_resource->second.get());

            // TODO Review with wan ip.
            if (tcp_sender_resource && remote_locator == tcp_sender_resource->channel()->remote_locator()) {
                // Look for an existing channel that matches this physical locator
                auto existing_channel = channel_resources_.find(remote_locator);
                // If the channel exists, check if the channel reference in the sender resource
                // needs to be updated with the found channel
                if (existing_channel != channel_resources_.end() &&
                    existing_channel->second != tcp_sender_resource->channel()) {
                    // Disconnect the old channel
                    tcp_sender_resource->channel()->disconnect();
                    tcp_sender_resource->channel()->clear();
                    // Update sender resource with new channel
                    tcp_sender_resource->channel() = existing_channel->second;
                }
                local_locator = tcp_sender_resource->channel()->locator();
                //TODO tcp_sender_resource->channel()存在，但是channel_resources_中没有该channel？
                return true;
            }
        }

        // At this point, if there no SenderResource to reuse, the channel was created for reception
        // (through an acceptor) or we have to create a new one.

        std::unique_lock<std::mutex> scopedLock(send_resources_mutex_);
        logDebug(RTCP,
                 "Called to OpenOutputChannel (physical: " << IPLocator::getPhysicalPort(remote_locator)
                                                           << "; logical: " << IPLocator::getLogicalPort(remote_locator)
                                                           << ") @ " << IPLocator::to_string(remote_locator));

        auto channel_resource = channel_resources_.find(remote_locator);

        std::shared_ptr<TCPChannelResource> channel;

        if (channel_resource != channel_resources_.end()) {
            channel = channel_resource->second;
            success = true;
        } else {
            // Create output channel
            logDebug(OpenOutputChannel,
                     "OpenOutputChannel (physical: " << IPLocator::getPhysicalPort(remote_locator) << "; "
                                                     << ") @ " << IPLocator::to_string(remote_locator));

            //TODO local locator need deal
            channel.reset(static_cast<TCPChannelResource*>(new TCPChannelResourceBasic(
                this, io_service_, remote_locator, remote_locator, configuration()->maxMessageSize)));
            success = channel->connect(channel);
        }
        if (success) {
            channel_resources_[remote_locator] = channel;
            send_resource_map[remote_locator] = std::shared_ptr<SenderResource>(new TCPSenderResource(*this, channel));
            local_locator = channel->locator();
            logInfo(RTCP, "OpenOutputChannel success, local_locator: " << IPLocator::to_string(local_locator)
                                                                       << " remote_locator: "
                                                                       << IPLocator::to_string(remote_locator));
        } else {
            logError(RTCP, "OpenOutputChannel failed, local_locator: " << IPLocator::to_string(local_locator)
                                                                       << " remote_locator: "
                                                                       << IPLocator::to_string(remote_locator));
        }
    } else {
        logError(RTCP, "OpenOutputChannel physical port is 0");
    }

    return success;
}

bool TCPTransportInterface::OpenInputChannel(const Locator& locator, std::shared_ptr<ReceiverResource> receiver,
                                             uint32_t /*maxMsgSize*/) {
    bool success = false;
    if (IsLocatorSupported(locator)) {
        uint16_t physicalPort = IPLocator::getPhysicalPort(locator);

        //如果没有配置listening_ports，则是client模式
        if (configuration()->listening_ports.empty()) {
            //查看channel的状态
            std::unique_lock<std::mutex> scopedLock(send_resources_mutex_);
            auto it = channel_resources_.begin();
            for (; it != channel_resources_.end(); ++it) {
                if (it->second->locator() == locator) {
                    if (it->second->connection_status() == TCPChannelResource::eConnectionStatus::eDisconnected) {
                        return success;
                    } else {
                        logInfo(RTCP, "OpenInputChannel channel_resource is connected");
                        break;
                    }
                }
            }

            if (it == channel_resources_.end()) {
                logError(RTCP, "OpenInputChannel channel_resource not found");
                return success;
            }
        }

        if (!is_input_port_open(physicalPort)) {
            success = true;
            {
                std::unique_lock<std::mutex> scopedLock(receiver_resources_mutex_);
                receiver_resources_[physicalPort] = std::move(receiver);
            }

            logDebug(RTCP, " OpenInputChannel (physical: " << IPLocator::getPhysicalPort(locator) << ")");
        }

        // (void)create_acceptor_socket(locator);
        for (uint16_t port : configuration()->listening_ports) {
            (void)port;
            Locator locator_tmp(LOCATOR_KIND_TCPv4, port);
            locator_tmp.set_address(locator);
            (void)create_acceptor_socket(locator_tmp);
        }
    }
    return success;
}

Locator TCPTransportInterface::RemoteToMainLocal(const Locator& remote) const {
    if (!IsLocatorSupported(remote)) {
        return false;
    }

    Locator mainLocal(remote);
    mainLocal.set_Invalid_Address();
    return mainLocal;
}

void TCPTransportInterface::perform_listen_operation(std::weak_ptr<TCPChannelResource> channel_weak) {
    Locator remote_locator;
    std::shared_ptr<TCPChannelResource> channel;
    uint16_t physicalPort(0);
    RTCPHeader& rtcp_header = configuration()->rtcp_header_;

    channel = channel_weak.lock();

    if (channel) {
        try {
            endpoint_to_locator(channel->remote_endpoint(), remote_locator);
        } catch (const std::exception& e) {
            logError(RTCP, "endpoint_to_locator failed: " << e.what());
        }
    }

    while (channel && (TCPChannelResource::eConnectionStatus::eConnecting < channel->connection_status())) {
        logDebug(RTCP, "TCPChannelResource status:" << channel->connection_status());
        // Blocking receive.
        unsigned char* buffer = channel->message_buffer();
        uint32_t length = 0;
        if (!Receive(channel, buffer, channel->buffer_size(), length, remote_locator, rtcp_header)) {
            continue;
        }

        // print the message
        if (TCPChannelResource::eConnectionStatus::eConnecting < channel->connection_status()) {
            physicalPort = IPLocator::getPhysicalPort(channel->locator());
            logInfo(RTCP, "physicalPort: " << physicalPort << " length: " << length);
            logInfo(RTCP, "length: " << length);
            std::unique_lock<std::mutex> scopedLock(receiver_resources_mutex_);
            auto it = receiver_resources_.find(physicalPort);

            if (it != receiver_resources_.end()) {
                std::shared_ptr<ReceiverResource> receiver = it->second;
                receiver->OnDataReceived(buffer, length, channel->locator(), remote_locator);
            } else {
                logInfo(RTPS_MSG_IN, "Received Message, but no receiver attached");
            }
        }
    }

    if (channel) {
        physicalPort = IPLocator::getPhysicalPort(channel->locator());
        std::unique_lock<std::mutex> scopedLock(receiver_resources_mutex_);
        auto it = receiver_resources_.find(physicalPort);

        if (it != receiver_resources_.end()) {
            std::shared_ptr<ReceiverResource> receiver = it->second;
            receiver->OnDisconnected(channel->locator(), remote_locator);
        } else {
            logInfo(RTPS_MSG_IN,
                    "Remote:" << IPLocator::to_string(remote_locator) << " Disconnected, but no receiver attached");
        }
    }
    logDebug(RTCP, "End PerformListenOperation " << remote_locator);
}

bool TCPTransportInterface::read_body(octet* receive_buffer, uint32_t, uint32_t* bytes_received,
                                      std::shared_ptr<TCPChannelResource>& channel, std::size_t body_size) {
    asio::error_code ec;

    *bytes_received = channel->read(receive_buffer, body_size, ec);

    if (ec) {
        logWarning(RTCP, "Error reading RTCP body: " << ec.message());
        return false;
    } else if (*bytes_received != body_size) {
        logError(RTCP, "Bad RTCP body size: " << *bytes_received << " (expected: " << body_size << ")");
        return false;
    }

    return true;
}

bool receive_header(std::shared_ptr<TCPChannelResource>& channel, RTCPHeader& rtcp_header, octet* receive_buffer,
                    uint32_t& receive_buffer_capacity, size_t& body_length, asio::error_code& ec) {

    if (rtcp_header.get_length() == 0 || rtcp_header.get_length() < RTCP_HEADER_MIN_SIZE ||
        receive_buffer_capacity < rtcp_header.get_length()) {
        return false;
    }

    octet* read_pos = receive_buffer;
    (void)memset(read_pos, 0, rtcp_header.get_length());
    size_t bytes_needed = RTCP_HEADER_ID_SIZE;

    logInfo(RTCP, "rtcp header: " << std::hex << std::setw(2) << std::setfill('0')
                                  << static_cast<int>(static_cast<unsigned char>(rtcp_header.get_rtcp()[0])) << " "
                                  << static_cast<int>(static_cast<unsigned char>(rtcp_header.get_rtcp()[1])) << " "
                                  << static_cast<int>(static_cast<unsigned char>(rtcp_header.get_rtcp()[2])) << " "
                                  << static_cast<int>(static_cast<unsigned char>(rtcp_header.get_rtcp()[3])));

    // Wait for sync
    while (bytes_needed > 0) {
        size_t bytes_read = channel->read(read_pos, bytes_needed, ec);
        if (bytes_read > 0) {
            logDebug(RTCP, "rtcp header: " << std::hex << std::setw(2) << std::setfill('0')
                                           << static_cast<int>(static_cast<unsigned char>(receive_buffer[0])) << " "
                                           << static_cast<int>(static_cast<unsigned char>(receive_buffer[1])) << " "
                                           << static_cast<int>(static_cast<unsigned char>(receive_buffer[2])) << " "
                                           << static_cast<int>(static_cast<unsigned char>(receive_buffer[3])));
            read_pos += bytes_read;
            bytes_needed -= bytes_read;
            if (0 == bytes_needed) {
                size_t skip =  // Text   Next possible match   Skip to next match
                    (receive_buffer[0] != rtcp_header.get_rtcp()[0]) ? 1 :  // X---   XRTCP                 1
                        (receive_buffer[1] != rtcp_header.get_rtcp()[1]) ? 1
                                                                         :  // RX--   RRTCP                 1
                        (receive_buffer[2] != rtcp_header.get_rtcp()[2]) ? 2
                                                                         :  // RTX-   RTRTCP                2
                        (receive_buffer[3] != rtcp_header.get_rtcp()[3]) ? 3
                                                                         : 0;  // RTCX   RTCRTCP               3

                if (skip) {
                    logInfo(RTCP, "skip: " << skip);
                    (void)memmove(receive_buffer, &receive_buffer[skip], RTCP_HEADER_ID_SIZE - skip);
                }

                read_pos -= skip;
                bytes_needed = skip;
            }
        } else if (ec) {
            return false;
        } else if (!channel->connection_status()) {
            logError(RTCP, "Channel disconnected while reading RTCP header");
            return false;
        }
    }

    bytes_needed = rtcp_header.get_length() - RTCP_HEADER_ID_SIZE;
    while (bytes_needed > 0) {
        size_t bytes_read = channel->read(read_pos, bytes_needed, ec);
        if (bytes_read > 0) {
            read_pos += bytes_read;
            bytes_needed -= bytes_read;
        } else if (ec) {
            return false;
        } else if (!channel->connection_status()) {
            return false;
        }
    }

    //TODO 解析rtcp_header
    body_length = *reinterpret_cast<uint32_t*>(receive_buffer + rtcp_header.get_offset());

    logInfo(RTCP, "rtcp_header body_length: " << body_length);

    if (!rtcp_header.is_used_default_header()) {
        // 如果使用自定义的header，需要将头部上传到用户态
        receive_buffer_capacity -= rtcp_header.get_length();
    }
    return true;
}

/**
 * On TCP, we must receive the header (14 Bytes) and then,
 * the rest of the message, whose length is on the header.
 * TCP Header is transparent to the caller, so receive_buffer
 * doesn't include it.
 * */
bool TCPTransportInterface::Receive(std::shared_ptr<TCPChannelResource>& channel, octet* receive_buffer,
                                    uint32_t receive_buffer_capacity, uint32_t& receive_buffer_size,
                                    Locator& remote_locator, RTCPHeader& rtcp_header) {
    (void)remote_locator;
    bool success = false;
    size_t body_size;

    try {
        success = true;

        // Read the header
        // octet header[TCPHEADER_SIZE];
        // TCPHeader tcp_header;
        asio::error_code ec;

        bool header_found = false;

        do {
            header_found = receive_header(channel, rtcp_header, receive_buffer, receive_buffer_capacity, body_size, ec);
        } while (!header_found && !ec && channel->connection_status());

        if (ec) {
            if (ec != asio::error::eof) {
                logWarning(DEBUG, "Failed to read TCP header: " << ec.message());
            }
            close_tcp_socket(channel);
            success = false;
        } else if (!channel->connection_status()) {
            logWarning(DEBUG, "Failed to read TCP header: channel disconnected while reading.");
            success = false;
            close_tcp_socket(channel);
        } else {

            // size_t body_size = tcp_header.length - static_cast<uint32_t>(TCPHeader::size());
            if (body_size > receive_buffer_capacity) {
                logError(RTCP_MSG_IN, "Size of incoming TCP message is bigger than buffer capacity: "
                                          << static_cast<uint32_t>(body_size) << " vs. " << receive_buffer_capacity
                                          << ". "
                                          << "The full message will be dropped.");
                success = false;
                // Drop the message
                size_t to_read = body_size;
                size_t read_block = receive_buffer_capacity;
                uint32_t readed;
                while (read_block > 0) {
                    (void)read_body(receive_buffer, receive_buffer_capacity, &readed, channel, read_block);
                    to_read -= readed;
                    read_block = (to_read >= receive_buffer_capacity) ? receive_buffer_capacity : to_read;
                }
            } else {
                if (!rtcp_header.is_used_default_header()) {
                    // 偏移receive_buffer的地址
                    receive_buffer += rtcp_header.get_length();
                }

                // logDebug(RTCP_MSG_IN, "Received RTCP MSG. Logical Port " << tcp_header.logical_port);
                success = read_body(receive_buffer, receive_buffer_capacity, &receive_buffer_size, channel, body_size);
                // Error message already shown by read_body method.
                if (!rtcp_header.is_used_default_header()) {
                    // 偏移receive_buffer的地址
                    receive_buffer_size += rtcp_header.get_length();
                }
            }
        }
    } catch (const asio::error_code& code) {
        if ((code == asio::error::eof) || (code == asio::error::connection_reset)) {
            // Close the channel
            logError(RTCP_MSG_IN, "ASIO [RECEIVE]: " << code.message());
            // channel->ConnectionLost();
            close_tcp_socket(channel);
        }
        success = false;
    } catch (const asio::system_error& error) {
        (void)error;
        // Close the channel
        logError(RTCP_MSG_IN, "ASIO SYSTEM_ERROR [RECEIVE]: " << error.what());
        // channel->ConnectionLost();
        close_tcp_socket(channel);
        success = false;
    }

    success = success && receive_buffer_size > 0;

    return success;
}

bool TCPTransportInterface::send(octet* send_buffer, uint32_t send_buffer_size,
                                 std::shared_ptr<TCPChannelResource>& channel,
                                 ertps::rtps::LocatorsIterator* destination_locators_begin,
                                 ertps::rtps::LocatorsIterator* destination_locators_end) {
    ertps::rtps::LocatorsIterator& it = *destination_locators_begin;

    bool ret = true;

    while (it != *destination_locators_end) {
        if (IsLocatorSupported(*it)) {
            ret &= send(send_buffer, send_buffer_size, channel, *it);
        }

        ++it;
    }

    return ret;
}

bool TCPTransportInterface::send(octet* send_buffer, uint32_t send_buffer_size,
                                 std::shared_ptr<TCPChannelResource>& channel, const Locator& remote_locator) {
    bool locator_mismatch = false;
    Locator channel_remote_locator = channel->remote_locator();
    octet* header = nullptr;
    size_t header_size = 0;
    size_t need_send_size = send_buffer_size;
    TCPHeader tcp_header;

    if (channel_remote_locator != remote_locator) {
        locator_mismatch = true;
    }

    if (locator_mismatch || send_buffer_size > configuration()->sendBufferSize) {

        logWarning(RTCP, "SEND [RTPS] Failed: Not connect: " << IPLocator::getLogicalPort(remote_locator)
                                                             << " @ IP: " << IPLocator::toIPv4string(remote_locator));
        return false;
    }

    if (send_buffer_size <= 0) {
        logWarning(RTCP, "send_buffer_size is too small:" << send_buffer_size);
        return false;
    }

    if (!configuration()->rtcp_header_.is_used_default_header()) {
        if (send_buffer_size < configuration()->rtcp_header_.get_length()) {
            logWarning(RTCP, "send_buffer_size is too small:" << send_buffer_size << "  header_size:"
                                                              << configuration()->rtcp_header_.get_length());
            return false;
        }
    }

    bool success = false;
    if (channel->connection_established()) {
        if (configuration()->rtcp_header_.is_used_default_header()) {
            tcp_header.set_length(send_buffer_size);
            header = tcp_header.address();
            header_size = tcp_header.size();
            need_send_size = send_buffer_size + header_size;
        }
        {
            asio::error_code ec;
            size_t sent = channel->send(header, header_size, send_buffer, send_buffer_size, ec);

            if (sent != need_send_size || ec) {
                logWarning(DEBUG, "Failed to send RTCP message (" << sent << " of " << need_send_size
                                                                  << " b): " << ec.message());
                success = false;
            } else {
                success = true;
            }
        }
    }

    return success;
}

bool TCPTransportInterface::async_send(octet* send_buffer, uint32_t send_buffer_size,
                                       std::shared_ptr<TCPChannelResource>& channel,
                                       ertps::rtps::LocatorsIterator* destination_locators_begin,
                                       ertps::rtps::LocatorsIterator* destination_locators_end,
                                       AsyncSendHandler handler) {
    ertps::rtps::LocatorsIterator& it = *destination_locators_begin;

    bool ret = true;

    while (it != *destination_locators_end) {
        if (IsLocatorSupported(*it)) {
            ret &= async_send(send_buffer, send_buffer_size, channel, *it, handler);
        }

        ++it;
    }

    return ret;
}

bool TCPTransportInterface::async_send(octet* send_buffer, uint32_t send_buffer_size,
                                       std::shared_ptr<TCPChannelResource>& channel, const Locator& remote_locator,
                                       AsyncSendHandler handler) {
    bool locator_mismatch = false;
    Locator channel_remote_locator = channel->remote_locator();
    octet* header = nullptr;
    size_t header_size = 0;
    TCPHeader tcp_header;
    if (channel_remote_locator != remote_locator) {
        locator_mismatch = true;
    }

    if (locator_mismatch || send_buffer_size > configuration()->sendBufferSize) {

        logWarning(RTCP, "SEND [RTPS] Failed: Not connect: " << IPLocator::getLogicalPort(remote_locator)
                                                             << " @ IP: " << IPLocator::toIPv4string(remote_locator));
        return false;
    }

    if (send_buffer_size <= 0) {
        logWarning(RTCP, "send_buffer_size is too small:" << send_buffer_size);
        return false;
    }

    if (!configuration()->rtcp_header_.is_used_default_header()) {
        if (send_buffer_size < configuration()->rtcp_header_.get_length()) {
            logWarning(RTCP, "send_buffer_size is too small:" << send_buffer_size << "  header_size:"
                                                              << configuration()->rtcp_header_.get_length());
            return false;
        }
    }

    bool success = false;
    if (channel->connection_established()) {
        // uint16_t physical_port = IPLocator::getPhysicalPort(remote_locator);
        if (configuration()->rtcp_header_.is_used_default_header()) {
            tcp_header.set_length(send_buffer_size);
            header = tcp_header.address();
            header_size = tcp_header.size();
        }
        {
            channel->async_send(header, header_size, send_buffer, send_buffer_size, std::move(handler),
                                configuration()->rtcp_header_.is_used_default_header());
            success = true;
        }
    }

    return success;
}

void TCPTransportInterface::SocketAccepted(std::shared_ptr<asio::ip::tcp::socket> socket, const Locator& locator,
                                           const asio::error_code& error) {
    Locator_t remote_locator;
    bool new_connection = false;
    std::shared_ptr<ReceiverResource> receiver;
    std::shared_ptr<TCPChannelResource> channel = nullptr;
    logDebug(RTCP, "SocketAccepted[alive_:" << alive_.load() << " error:" << error.value() << "]");
    if (alive_.load()) {
        if (!error.value()) {
            // Store the new connection.
            try {
                endpoint_to_locator(socket->remote_endpoint(), remote_locator);
            } catch (const std::exception& e) {
                logError(RTCP, "endpoint_to_locator failed: " << e.what());
                return;
            }

            {
                std::unique_lock<std::mutex> scopedLock(acceptors_mutex_);
                std::shared_ptr<TCPAcceptor> acceptor = acceptors_[locator];
                if (acceptor == nullptr) {
                    logDebug(RTCP, "locator:" << IPLocator::to_string(locator) << " in acceptors_ is removed");
                    return;
                }
            }
            channel = std::shared_ptr<TCPChannelResource>(new TCPChannelResourceBasic(
                this, io_service_, std::move(socket), locator, remote_locator, configuration()->maxMessageSize));

            {
                std::unique_lock<std::mutex> scopedLock(send_resources_mutex_);
                channel_resources_[channel->remote_locator()] = channel;
                logDebug(RTCP, "SocketAccepted[channel_resources_ size:" << channel_resources_.size() << "]");
            }

            //通知上层有新的连接
            uint16_t physicalPort = IPLocator::getPhysicalPort(channel->locator());
            {
                std::unique_lock<std::mutex> scopedLock(receiver_resources_mutex_);
                auto it = receiver_resources_.find(physicalPort);

                if (it != receiver_resources_.end()) {
                    receiver = it->second;
                    new_connection = true;
                    logInfo(RTPS_MSG_IN, "Received Message, return OnNewAccepted");
                } else {
                    logInfo(RTPS_MSG_IN, "Received Message, but no receiver attached");
                }
            }

            if (new_connection) {
                receiver->OnNewAccepted(channel->locator(), remote_locator);
            }

            channel->set_options(configuration());
            std::weak_ptr<TCPChannelResource> channel_weak_ptr = channel;
            (void)channel->change_status(TCPChannelResource::eConnectionStatus::eEstablished);
            channel->thread(std::thread(&TCPTransportInterface::perform_listen_operation, this, channel_weak_ptr));
            try {
                logDebug(RTCP, " Accepted connection (local: " << IPLocator::to_string(locator)
                                                               << ", remote: " << channel->remote_endpoint().address()
                                                               << ":" << channel->remote_endpoint().port() << ")");
            } catch (const asio::system_error& e) {
                logWarning(RTCP, "catch err: " << e.what());
            }
        } else {
            logDebug(RTCP, " Accepting connection (" << error.message() << ")");
            std::this_thread::sleep_for(std::chrono::milliseconds(200));  // Wait a little to accept again.
        }

        if (error.value() != eSocketErrorCodes::eConnectionAborted)  // Operation Aborted
        {
            std::unique_lock<std::mutex> scopedLock(acceptors_mutex_);
            std::shared_ptr<TCPAcceptor> acceptor = acceptors_[locator];
            if (acceptor != nullptr) {
                TCPAcceptorBasic* TcpAccept = dynamic_cast<TCPAcceptorBasic*>(acceptor.get());
                if (nullptr != TcpAccept) {
                    TcpAccept->accept(this);
                }
            } else {
                logDebug(RTCP, "SocketAccepted[acceptor == nullptr] channel disable");
                std::unique_lock<std::mutex> scopedLock(send_resources_mutex_);
                logDebug(RTCP, "send_resources_mutex_ get lock success");
                if (channel && channel.get() != nullptr) {
                    // channel->change_status(TCPChannelResource::eConnectionStatus::eDisconnected);
                    channel->disconnect();
                    channel->clear();
                    logDebug(RTCP, "SocketAccepted clear channel status" << channel->connection_status());
                } else {
                    logDebug(RTCP, "SocketAccepted channel is nullptr");
                }
            }
        }
    }
}

void TCPTransportInterface::SocketConnected(const std::weak_ptr<TCPChannelResource>& channel_weak_ptr,
                                            const asio::error_code& error) {
    if (alive_.load()) {
        auto channel = channel_weak_ptr.lock();

        if (channel) {
            if (!error) {

                //创建收包资源
                // OpenInputChannel(channel->locator());
                if (TCPChannelResource::eConnectionStatus::eDisconnected < channel->connection_status()) {
                    (void)channel->change_status(TCPChannelResource::eConnectionStatus::eEstablished);
                    channel->set_options(configuration());
                    channel->thread(
                        std::thread(&TCPTransportInterface::perform_listen_operation, this, channel_weak_ptr));
                }
            } else {
                channel->disconnect();
            }
        }
    }
}

bool TCPTransportInterface::getDefaultMetatrafficMulticastLocators(LocatorList& locators,
                                                                   uint32_t metatraffic_multicast_port) const {
    // TCP doesn't have multicast support
    (void)locators;
    (void)metatraffic_multicast_port;
    return true;
}

bool TCPTransportInterface::getDefaultMetatrafficUnicastLocators(LocatorList& locators,
                                                                 uint32_t metatraffic_unicast_port) const {
    (void)locators;
    (void)metatraffic_unicast_port;
    return true;
}

bool TCPTransportInterface::getDefaultUnicastLocators(LocatorList& locators, uint32_t unicast_port) const {
    (void)locators;
    (void)unicast_port;
    return true;
}

bool TCPTransportInterface::fillMetatrafficMulticastLocator(Locator& locator,
                                                            uint32_t metatraffic_multicast_port) const {
    (void)locator;
    (void)metatraffic_multicast_port;
    // TCP doesn't have multicast support
    return true;
}

bool TCPTransportInterface::fillMetatrafficUnicastLocator(Locator& locator, uint32_t metatraffic_unicast_port) const {
    (void)locator;
    (void)metatraffic_unicast_port;
    return true;
}

bool TCPTransportInterface::configureInitialPeerLocator(Locator& locator, const PortParameters& port_params,
                                                        uint32_t domainId, LocatorList& list) const {
    (void)locator;
    (void)port_params;
    (void)domainId;
    (void)list;
    return true;
}

bool TCPTransportInterface::fillUnicastLocator(Locator& locator, uint32_t well_known_port) const {
    (void)locator;
    (void)well_known_port;
    return true;
}

void TCPTransportInterface::update_network_interfaces() {
    // TODO(jlbueno)
}

}  // namespace transport
}  // namespace vbs
